package cn.itcast.streaming.source;

import cn.itcast.streaming.bean.RuleInfoModel;
import cn.itcast.streaming.utils.JDBCUtil;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

//读取自定义告警规则数据并返回javabean
public class MysqlCustomRuleInfoSource extends RichSourceFunction<ArrayList<RuleInfoModel>> {

    private static Logger logger = LoggerFactory.getLogger(MysqlElectricFenceResultSource.class);
    private ParameterTool parameterTool; //全局参数对象
    private Connection conn;
    private Statement statement;
    private Boolean isRunning = true;
    private String sql = "select rule_name,\n" +
            "       alarm_param1_field,operator1,rule_symbol1,alarm_threshold1,logical_symbol,\n" +
            "       alarm_param2_field,rule_symbol2,alarm_threshold2,operator2,logical_symbol2,\n" +
            "       alarm_param3_field,rule_symbol3,alarm_threshold3,\n" +
            "       alarm_param4_field,rule_symbol4,alarm_threshold4,alarm_frame,\n" +
            "       monitor_type_id,series_name,series_code,model_code,model_name,\n" +
            "       province,city,vins,monitor_type_name,task.id\n" +
            "from t_alarm_rule rule\n" +
            "inner join t_monitor_task task on rule.id = task.alarm_rule_id\n" +
            "inner join t_monitor_type type on task.monitor_type_id = type.id\n" +
            "where rule.STATUS=1 and task.STATUS=1;";

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        parameterTool = (ParameterTool)getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
        conn = JDBCUtil.getConnection();
        statement = conn.createStatement();
    }

    @Override
    public void close() throws Exception {
        super.close();
        if(statement!=null && !statement.isClosed()) statement.close();
        if(conn!=null && !conn.isClosed()) conn.close();
    }

    @Override
    public void run(SourceContext<ArrayList<RuleInfoModel>> sourceContext) throws Exception {
        ArrayList<RuleInfoModel> ruleInfoModelList = new ArrayList<>();
        while (isRunning) {
            ruleInfoModelList.clear();
            ResultSet resultSet = statement.executeQuery(sql);
            while (resultSet.next()) {
                RuleInfoModel ruleInfoModel = new RuleInfoModel(
                        resultSet.getString("rule_name"),
                        resultSet.getString("alarm_frame"),
                        resultSet.getString("monitor_type_id"),
                        resultSet.getString("series_name"),
                        resultSet.getString("series_code"),
                        resultSet.getString("model_code"),
                        resultSet.getString("model_name"),
                        resultSet.getString("province"),
                        resultSet.getString("city"),
                        resultSet.getString("vins"),
                        resultSet.getString("monitor_type_name"),
                        resultSet.getString("id"),
                        resultSet.getString("alarm_param1_field"),
                        resultSet.getString("operator1"),
                        resultSet.getString("rule_symbol1"),
                        resultSet.getString("alarm_threshold1"),
                        resultSet.getString("logical_symbol"),
                        resultSet.getString("alarm_param2_field"),
                        resultSet.getString("rule_symbol2"),
                        resultSet.getString("alarm_threshold2"),
                        resultSet.getString("operator2"),
                        resultSet.getString("logical_symbol2"),
                        resultSet.getString("alarm_param3_field"),
                        resultSet.getString("rule_symbol3"),
                        resultSet.getString("alarm_threshold3"),
                        resultSet.getString("alarm_param4_field"),
                        resultSet.getString("rule_symbol4"),
                        resultSet.getString("alarm_threshold4")
                );
                ruleInfoModelList.add(ruleInfoModel);
            }
            sourceContext.collect(ruleInfoModelList);

            //判断是否存在规则数据
            if(ruleInfoModelList.isEmpty()){
                logger.info("从自定义规则表中获取的数据是空的");
            }else {
                logger.info("从自定义规则表中获取的数据："+ruleInfoModelList);
            }

            TimeUnit.SECONDS.sleep(Long.parseLong(parameterTool.getRequired("ruleinfo.millionseconds")));
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
